package org.skyemoon.hbase.hbasetest.service.impl;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchPhraseQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.skyemoon.hbase.hbasetest.common.elasticsearch.ElasticsearchConnect;
import org.skyemoon.hbase.hbasetest.dao.entity.FlightImageDO;
import org.skyemoon.hbase.hbasetest.dto.domain.FlightDTO;
import org.skyemoon.hbase.hbasetest.dto.domain.FlightImageDTO;
import org.skyemoon.hbase.hbasetest.dto.domain.ImageDTO;
import org.skyemoon.hbase.hbasetest.dto.req.FlightImageReqDTO;
import org.skyemoon.hbase.hbasetest.dto.resp.FlightImageRespDTO;
import org.skyemoon.hbase.hbasetest.dto.resp.FlightRespDTO;
import org.skyemoon.hbase.hbasetest.service.HbaseService;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Base64;
import java.util.Date;
import java.util.List;

import static org.skyemoon.hbase.hbasetest.common.constant.HbaseConnectionConstant.ROW_KEY_PREFIX;
import static org.skyemoon.hbase.hbasetest.common.constant.HbaseTableConstant.*;
import static org.skyemoon.hbase.hbasetest.toolkit.Date2String.convertDate2String;

/**
 * HbaseService实现类
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class HbaseServiceImpl implements HbaseService {

    private final Connection connection;

    @Override
    public Boolean saveInfo(FlightImageDTO requestParam) throws IOException {
        System.out.println("-----开始写入数据-----");
        long start = System.currentTimeMillis();

        if (connection == null)
            return false;

        List<ImageDTO> imageList = requestParam.getImageList();
        byte[] flightTimeBytes = convertDate2String(requestParam.getFlightTime()).getBytes();
        byte[] flightTypeBytes = requestParam.getFlightType().getBytes();
        byte[] flightNumBytes = requestParam.getFlightNum().getBytes();
        byte[] flightTaskBytes = requestParam.getFlightTask().getBytes();

        List<FlightDTO> flightDTOList = new ArrayList<>();

        //TODO 添加大文件直接写入HDFS的代码，Hbase中存dfs路径

        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(TABLE_NAME));
        params.writeBufferSize(10 * 1024 * 1024);
        BufferedMutator mutator = connection.getBufferedMutator(params);
        try {
            for (ImageDTO image : imageList) {
                // rowKey设计：Long.MAX_VALUE-Time _ Prefix : imageName
                long createTimeReverse = Long.MAX_VALUE - System.currentTimeMillis();
                String rowKey = createTimeReverse + "_" + ROW_KEY_PREFIX + image.getImageName();
                Put put = new Put(rowKey.getBytes());
                put.addColumn(FAMILY_META.getBytes(), FLIGHT_TYPE.getBytes(), flightTypeBytes);
                put.addColumn(FAMILY_META.getBytes(), FLIGHT_TIME.getBytes(), flightTimeBytes);
                put.addColumn(FAMILY_META.getBytes(), FLIGHT_NUM.getBytes(), flightNumBytes);
                put.addColumn(FAMILY_META.getBytes(), FLIGHT_TASK.getBytes(), flightTaskBytes);
                put.addColumn(FAMILY_IMAGE.getBytes(), IMAGE_NAME.getBytes(), image.getImageName().getBytes());
                put.addColumn(FAMILY_IMAGE.getBytes(), IMAGE_TYPE.getBytes(), image.getImageType().getBytes());
                put.addColumn(FAMILY_IMAGE.getBytes(), IMAGE_SIZE.getBytes(), image.getImageSize().toString().getBytes());
                put.addColumn(FAMILY_IMAGE.getBytes(), IMAGE.getBytes(), image.getImage());
                mutator.mutate(put);
                FlightDTO flightDTO = FlightDTO.builder()
                        .flightType(requestParam.getFlightType())
                        .flightTask(requestParam.getFlightTask())
                        .flightTime(requestParam.getFlightTime())
                        .flightNum(requestParam.getFlightNum())
                        .rowKey(rowKey)
                        .build();
                flightDTOList.add(flightDTO);
            }
            mutator.flush();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            mutator.close();
        }

        ElasticsearchConnect.connect(client -> {
            BulkRequest request = new BulkRequest();
            flightDTOList.forEach(each -> {
                request.add(new IndexRequest().index("flight_images").source(XContentType.JSON,
                        "flight_type", each.getFlightType(), "flight_task", each.getFlightTask(),
                        "flight_num", each.getFlightNum(), "flight_time", each.getFlightTime(),
                        "row_key", each.getRowKey()
                ));
            });
            client.bulk(request, RequestOptions.DEFAULT);
            return null;
        });

        long stop = System.currentTimeMillis();
        System.out.println("-----写入数据时间:"+(stop - start)*1.0/1000 + "s-----");
        return true;
    }

    public List<FlightRespDTO> queryImageByInfo(FlightImageReqDTO flightImageReqDTO) {
        // TODO es查询size只有10，等待解决
        List<FlightRespDTO> result = new ArrayList<>();
        List<String> rowKeysInEs = queryRowKeyInEs(flightImageReqDTO);
        System.out.println(rowKeysInEs);
        rowKeysInEs.forEach(each -> {
            try {
                FlightImageDO flightImageDO = selectByRowKey(each);
                FlightRespDTO flightRespDTO = FlightRespDTO.builder()
                        .flightType(flightImageDO.getFlightType())
                        .flightTime(flightImageDO.getFlightTime())
                        .flightNum(flightImageDO.getFlightNum())
                        .flightTask(flightImageDO.getFlightTask())
                        .build();
                result.add(flightRespDTO);
                // TODO 文件写入被拒绝访问，初步判定是没加文件名的原因，需要使用原始文件名
//                if (flightImageReqDTO.getPath() != null) {
//                    File file = new File(flightImageReqDTO.getPath());
//                    try {
//                        FileOutputStream fos = new FileOutputStream(file);
//                        fos.write(flightImageDO.getImage());
//                        fos.close();
//                    } catch (IOException e) {
//                        e.printStackTrace();
//                    }
//                }
            } catch (IOException | ParseException e) {
                throw new RuntimeException(e);
            }
        });
        return result;
    }

    public List<String> queryRowKeyInEs(FlightImageReqDTO flightImageReqDTO) {
        Object rowKeyList = ElasticsearchConnect.connect(client -> {
            List<String> rowKeys = new ArrayList<>();
            SearchRequest request = new SearchRequest();
            request.indices("flight_images");
            // 构建查询的请求体
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            if (flightImageReqDTO.getFlightType() != null) {
                MatchPhraseQueryBuilder query1 = QueryBuilders.matchPhraseQuery("flight_type", flightImageReqDTO.getFlightType());
                boolQueryBuilder.must(query1);
            }
            if (flightImageReqDTO.getFlightTime() != null) {
                MatchPhraseQueryBuilder query2 = QueryBuilders.matchPhraseQuery("flight_time", flightImageReqDTO.getFlightTime());
                boolQueryBuilder.must(query2);
            }
            if (flightImageReqDTO.getFlightNum() != null) {
                MatchPhraseQueryBuilder query3 = QueryBuilders.matchPhraseQuery("flight_num", flightImageReqDTO.getFlightNum());
                boolQueryBuilder.must(query3);
            }
            if (flightImageReqDTO.getFlightTask() != null) {
                MatchPhraseQueryBuilder query4 = QueryBuilders.matchPhraseQuery("flight_task", flightImageReqDTO.getFlightTask());
                boolQueryBuilder.must(query4);
            }
            sourceBuilder.query(boolQueryBuilder);
            request.source(sourceBuilder);
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 查询匹配
            SearchHits hits = response.getHits();
            for (SearchHit hit : hits) {
                String rowKey = (String) hit.getSourceAsMap().get("row_key");
                rowKeys.add(rowKey);
            }
            return rowKeys;
        });
        return (List<String>) rowKeyList;
    }

    @Override
    public FlightImageDO selectByRowKey(String rowKey) throws IOException, ParseException {
        if (connection == null)
            return null;
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        String flightType = new String(queryObjectByRowKey(table, rowKey, FAMILY_META, FLIGHT_TYPE));
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date flightTime = formatter.parse(new String(queryObjectByRowKey(table, rowKey, FAMILY_META, FLIGHT_TIME)));
        String flightNum = new String(queryObjectByRowKey(table, rowKey, FAMILY_META, FLIGHT_NUM));
        String flightTask = new String(queryObjectByRowKey(table, rowKey, FAMILY_META, FLIGHT_TASK));
        String imageName = new String(queryObjectByRowKey(table, rowKey, FAMILY_IMAGE, IMAGE_NAME));
        Long imageSize = Long.parseLong(new String(queryObjectByRowKey(table, rowKey, FAMILY_IMAGE, IMAGE_SIZE)));
        String imageType = new String(queryObjectByRowKey(table, rowKey, FAMILY_IMAGE, IMAGE_TYPE));
        byte[] image = queryObjectByRowKey(table, rowKey, FAMILY_IMAGE, IMAGE);
        table.close();
        return FlightImageDO.builder()
                .flightType(flightType)
                .flightTime(flightTime)
                .flightNum(flightNum)
                .flightTask(flightTask)
                .imageType(imageType)
                .imageSize(imageSize)
                .imageName(imageName)
                .image(image)
                .build();
    }

    public byte[] queryObjectByRowKey(Table table, String rowKey, String columnFamily, String columnName) {
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
        get.getMaxVersions();
        Result result = null;
        try {
            result = table.get(get);
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (result != null)
            return result.getValue(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName));
        return null;
    }
}
